package com.lanlian.cayman.rabbitmq;

import android.os.Handler;
import android.util.Log;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;

/**
 * <pre>
 *     author : huxiaobo
 *     e-mail : huxiaobo@xlanlian.com
 *     date   : 2018/8/20 15:16
 *     desc   : MQ消费者
 *     version: 1.0
 * </pre>
 */
public class MQConsumer extends MQConnector {

    private static final String TAG = "MQConsumer";

    private MQCallback mCallback;
    private ExecutorService executor;
    private String mQueueName;
    private String mExchange;
    private String mRoutingKey;
    private BuiltinExchangeType mExchangeType = BuiltinExchangeType.FANOUT;

    private Handler mCallbackHandler = new Handler();

    /**
     * 消息接受监听
     */
    private MQConsumerListener mqConsumerListener;

    public interface MQConsumerListener {
        void onMessageReceived(byte[] bytes);
    }

    public static MQConsumer createInstance(MQFactory factory, MQCallback callback, String exchange, String routingKey) {
        return new MQConsumer(factory.getHostName(), factory.getUsername(), factory.getPassword(),
                factory.getPort(), exchange, routingKey, factory.getExecutor(), callback
        );
    }

    private MQConsumer(String host, String username, String password, int port, String exchange,
                       String routingKey, ExecutorService executor, MQCallback callback) {
        super(host, username, password, port);
        this.mCallback = callback;
        this.mExchange = exchange;
        this.mRoutingKey = routingKey;
        this.executor = executor;
    }

    public void setMessageListner(MQConsumerListener listener) {
        mqConsumerListener = listener;
    }

    @Override
    protected ShutdownListener createShutDownListener() {
        return new ShutdownListener() {
            @Override
            public void shutdownCompleted(ShutdownSignalException cause) {
                String errorMessage = cause.getMessage() == null ? "cunsumer connection was shutdown" :
                        "consumer " + cause.getMessage();
                mCallback.onMQConnectionClosed(errorMessage);
            }
        };
    }

    private class CloseThread extends Thread {
        @Override
        public void run() {
            try {
                closeMQConnection();
            } catch (Exception e) {
                sendBackErrorMessage(e);
                e.printStackTrace();
            }
        }
    }

    /**
     * 关闭消费者资源
     */
    public void close() {
        if (mIsSubscribe) {
            mIsSubscribe = false;
            new CloseThread().start();
        }
    }

    private boolean mIsSubscribe;

    /**
     * 订阅消息
     */
    public void subscribe() {
        Log.d(TAG, "subscribe");
        if (executor.isShutdown()) {
            return;
        }
        executor.execute(new Runnable() {
            @Override
            public void run() {
                synchronized (this) {
                    try {
                        if (!mIsSubscribe) {
                            mIsSubscribe = true;
                            initConnection();
                            initChanenel();
                            if (mChannel == null) {
                                return;
                            }
                            mChannel.basicQos(1);
                            AMQP.Queue.DeclareOk q = mChannel.queueDeclare();
                            mQueueName = q.getQueue();
                            mChannel.queueBind(mQueueName, mExchange, mRoutingKey);
                            Consumer consumer = new DefaultConsumer(mChannel) {
                                @Override
                                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                                        throws IOException {
                                    super.handleDelivery(consumerTag, envelope, properties, body);
                                    mCallbackHandler.post(new Runnable() {
                                        @Override
                                        public void run() {
                                            if (mqConsumerListener != null) {
                                                mqConsumerListener.onMessageReceived(body);
                                            }
                                        }
                                    });
                                }
                            };
                            mChannel.basicConsume(mQueueName, true, consumer);
                        }
                    } catch (ConsumerCancelledException | ShutdownSignalException |
                            IOException | TimeoutException e) {
                        sendBackErrorMessage(e);
                        try {
                            Thread.sleep(5000);
                            mIsSubscribe = false;
                            subscribe();
                        } catch (InterruptedException e1) {
                            e1.printStackTrace();
                        }
                    }
                }
            }
        });
    }

    /**
     * 发生错误信息
     *
     * @param e 异常信息
     */
    private void sendBackErrorMessage(Exception e) {
        final String errorMessage = e.getMessage() == null ? e.toString() : e.getMessage();
        mCallbackHandler.post(new Runnable() {
            @Override
            public void run() {
                mCallback.onMQConnectionFailure(errorMessage);
            }
        });
    }

    /**
     * 初始化连接
     */
    private void initConnection() throws IOException, TimeoutException {
        if (!isConnected()) {
            createConnection();
        }
    }

    /**
     * 初始化通道
     */
    private void initChanenel() {
        if (!isChannelAvailable()) {
            createChannel();
        }
    }

    public String getExchange() {
        return mExchange;
    }

    public void setExchange(String mExchange) {
        this.mExchange = mExchange;
    }

    public String getRoutingkey() {
        return mRoutingKey;
    }

    public void setRoutingkey(String mRoutingKey) {
        this.mRoutingKey = mRoutingKey;
    }

    public String getQueueName() {
        return mQueueName;
    }

    public void setQueueName(String mQueueName) {
        this.mQueueName = mQueueName;
    }

    public BuiltinExchangeType getExchangeType() {
        return mExchangeType;
    }

    public void setExchangeType(BuiltinExchangeType exchangeType) {
        this.mExchangeType = exchangeType;
    }

}
